package com.nbcio.iot.protocol.tcp;

import com.nbcio.iot.config.IotServerProperties;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LineBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.util.CharsetUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.BiConsumer;

/**
 * TCP服务器管理器
 * 支持多客户端连接，接收TCP数据并转发到数据处理流程
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class TcpServerManager {

    private final AtomicBoolean running = new AtomicBoolean(false);
    private final ConcurrentHashMap<String, Channel> clientConnections = new ConcurrentHashMap<>();
    private EventLoopGroup bossGroup;
    private EventLoopGroup workerGroup;
    private Channel serverChannel;
    private BiConsumer<String, String> messageListener;
    private int serverPort = 9999; // 默认端口
    private int bufferSize = 4096; // 可通过配置覆盖

    private final IotServerProperties serverProperties;

    public void setMessageListener(BiConsumer<String, String> listener) {
        this.messageListener = listener;
    }

    public void setServerPort(int port) {
        this.serverPort = port;
    }

    @PostConstruct
    public void start() {
        // 读取配置
        if (serverProperties != null && serverProperties.getTcpServer() != null) {
            this.serverPort = serverProperties.getTcpServer().getPort();
            this.bufferSize = serverProperties.getTcpServer().getBufferSize();
            boolean enabled = serverProperties.getTcpServer().isEnabled();
            if (!enabled) {
                log.info("[TCP-SERVER] 配置为关闭，跳过启动 (port={})", serverPort);
                return;
            }
        }

        if (running.compareAndSet(false, true)) {
            bossGroup = new NioEventLoopGroup(1);
            workerGroup = new NioEventLoopGroup();
            try {
                ServerBootstrap b = new ServerBootstrap();
                b.group(bossGroup, workerGroup)
                 .channel(NioServerSocketChannel.class)
                 .option(ChannelOption.SO_BACKLOG, 1024)
                 .childOption(ChannelOption.TCP_NODELAY, true)
                 .childOption(ChannelOption.SO_KEEPALIVE, true)
                 .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                 .childHandler(new ChannelInitializer<SocketChannel>() {
                     @Override
                     protected void initChannel(SocketChannel ch) {
                         ChannelPipeline p = ch.pipeline();
                         p.addLast(new LineBasedFrameDecoder(bufferSize));
                         p.addLast(new StringDecoder(CharsetUtil.UTF_8));
                         p.addLast(new SimpleChannelInboundHandler<String>() {
                             private String clientId(ChannelHandlerContext ctx) {
                                 return ctx.channel().remoteAddress().toString().replaceFirst("^/", "");
                             }
                             @Override
                             public void channelActive(ChannelHandlerContext ctx) {
                                 String id = clientId(ctx);
                                 clientConnections.put(id, ctx.channel());
                                 log.info("[TCP-SERVER] 新客户端连接: {}", id);
                             }
                             @Override
                             protected void channelRead0(ChannelHandlerContext ctx, String msg) {
                                 String id = clientId(ctx);
                                 if (msg == null || msg.trim().isEmpty()) return;
                                 log.info("[TCP-SERVER] 收到客户端 {} 数据: {}", id, msg);
                                 if (messageListener != null) {
                                     messageListener.accept("tcp/" + id, msg);
                                 }
                             }
                             @Override
                             public void channelInactive(ChannelHandlerContext ctx) {
                                 String id = clientId(ctx);
                                 clientConnections.remove(id);
                                 log.info("[TCP-SERVER] 客户端 {} 已断开", id);
                             }
                             @Override
                             public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
                                 String id = clientId(ctx);
                                 log.warn("[TCP-SERVER] 客户端 {} 异常: {}", id, cause.getMessage());
                                 ctx.close();
                             }
                         });
                     }
                 });

                ChannelFuture f = b.bind(serverPort).sync();
                serverChannel = f.channel();
                log.info("[TCP-SERVER] TCP服务器(Netty)启动，监听端口: {}", serverPort);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                log.error("[TCP-SERVER] 启动中断", e);
                running.set(false);
                shutdownGroups();
            } catch (Exception e) {
                log.error("[TCP-SERVER] 启动失败，端口: {}", serverPort, e);
                running.set(false);
                shutdownGroups();
            }
        }
    }

    @PreDestroy
    public void stop() {
        if (running.compareAndSet(true, false)) {
            log.info("[TCP-SERVER] 正在停止TCP服务器...");
            try {
                if (serverChannel != null) {
                    serverChannel.close().syncUninterruptibly();
                }
            } finally {
                clientConnections.clear();
                shutdownGroups();
            }
            log.info("[TCP-SERVER] TCP服务器已停止");
        }
    }

    private void shutdownGroups() {
        if (bossGroup != null) bossGroup.shutdownGracefully();
        if (workerGroup != null) workerGroup.shutdownGracefully();
    }

    /**
     * 向指定客户端发送数据
     */
    public boolean sendToClient(String clientId, String message) {
        Channel ch = clientConnections.get(clientId);
        if (ch != null && ch.isActive()) {
            ch.writeAndFlush(message + "\n");
            log.info("[TCP-SERVER] 向客户端 {} 发送数据: {}", clientId, message);
            return true;
        }
        return false;
    }

    /**
     * 广播消息给所有客户端
     */
    public void broadcast(String message) {
        clientConnections.keySet().forEach(clientId -> sendToClient(clientId, message));
    }

    /**
     * 获取当前连接的客户端数量
     */
    public int getClientCount() {
        return clientConnections.size();
    }

    /**
     * 获取所有客户端ID
     */
    public java.util.Set<String> getClientIds() {
        return clientConnections.keySet();
    }
}

